package test.org.offsync.nio

import java.nio.channels.CancelledKeyException
import java.nio.channels.ClosedChannelException
import java.nio.channels.SelectionKey
import java.nio.channels.{SelectionKey => JavaSelectionKey}
import org.junit.runner.RunWith
import org.offsync.async.DeadlineTimer
import org.offsync.async.AsyncSelector
import org.offsync.async.AsyncServerSocket
import org.offsync.async.AsyncSocket
import org.offsync.rich._
import org.offsync.util.DisposedException
import org.offsync.util.Duration
import org.offsync.util.Slot
import org.offsync.util.Time
import org.offsync.util.Trace.trace
import org.offsync.util.Window
import scala.collection.mutable.ListBuffer
import test.util.AssertionsCollector
import test.util.Timeline
import test.util.Timeline._
import java.net.InetSocketAddress
import org.junit.Test
import org.junit.Assert
import java.util.concurrent.CyclicBarrier
import java.util.concurrent.TimeUnit
import java.util.concurrent.TimeoutException
import java.util.concurrent.Semaphore

class SpecSocketChannel {
  def startThread(f: => Unit): Thread = {
    val serverThread = new Thread() {
      override def run(): Unit = f
    }
    
    serverThread.start()
    
    return serverThread
  }
  
  val portRange = 2000 to 3000

  @Test
  def testAllowExecutionInSelectorThread(): Unit = {
    val selector = new AsyncSelector()
    var result0: Option[Throwable] = None
    val semaphore = new Semaphore(0)
    AsyncServerSocket.open("localhost", portRange) { completion =>
      val (serverSocket, port) = completion()
      serverSocket.accept(selector) { completion =>
        try {
          completion()
        } catch {
          case e: DisposedException => result0 = Some(e)
        }
        false
      }
      selector.execute { semaphore.release() }
      semaphore.acquire()
      selector.execute { semaphore.release() }
      semaphore.acquire()
    }

    Thread.sleep(400)
    
    selector.dispose()
    assert(result0.get.getClass().equals(classOf[DisposedException]))
  }

  @Test
  def testAccept(): Unit = {
    val selector = new AsyncSelector()
    val portSlot = new Slot[Int]
    val connected = new Slot[Boolean]
    val complete = new Slot[Boolean]
    
    val serverThread = startThread {
      AsyncServerSocket.open("localhost", portRange) { completion =>
        val (listenChannel, port) = completion()
        portSlot.set(port)
        listenChannel.accept(selector) { completion =>
          try {
            try {
              completion()
              connected.get
            } finally {
              complete.set(true)
            }
          } catch {
            case e => {
              println(e)
              e.printStackTrace(System.out)
              complete.set(false)
            }
          }
          false
        }
        complete.get
      }
    }
    
    AsyncSocket.open("localhost", portSlot.get).using { clientChannel =>
      connected.set(true)
      complete.get
    }
    
    serverThread.join()
    selector.dispose()
    assert(complete.get == true)
  }
  
  @Test
  def testReadWrite(): Unit = {
    import Default._
    var requireThat = new AssertionsCollector
    val selector = new AsyncSelector()
    
    val deadlineTimer = new DeadlineTimer()
    var result0 = false
    val timeline = new Timeline
    var runCount = 0
    val portSlot = new Slot[Integer](None)
    
    val serverThread = startThread {
      AsyncServerSocket.open("localhost", portRange) { completion =>
        val (serverChannel, port) = completion()
        portSlot.set(port)
        val complete = new Slot[Boolean](None)
        serverChannel.accept(selector) { getSocket =>
          try {
            for (sessionChannel <- getSocket()) {
              val window = Window to (new Array[Byte](100))
              def performSessionRead(): Unit = {
                sessionChannel.readSome(selector, window) { getBytesRead =>
                  try {
                    val bytesRead = getBytesRead()
                    if (bytesRead >= 0) {
                      sessionChannel.writeSome(selector, window.until(bytesRead)) { getBytesWritten =>
                        try {
                          val bytesWritten = getBytesWritten()
                          requireThat { assert(bytesWritten == 11 || bytesWritten == 4 || bytesWritten == 15) }
                          performSessionRead()
                        } catch {
                          case e: DisposedException => {
                            complete.set(true)
                            e.printStackTrace(System.out)
                          }
                          case e => {
                            complete.set(true)
                            println("unexpected exception: " + e)
                            assert(false)
                          }
                        }
                      }
                    } else {
                      complete.set(true)
                    }
                  } catch {
                    case e: DisposedException => {
                      println("--> in read server channel disposed *************")
                      complete.set(true)
                      e.printStackTrace(System.out)
                    }
                    case e => {
                      complete.set(true)
                      e.printStackTrace(System.out)
                      println("unexpected exception: " + e)
                      assert(false)
                    }
                  }
                }
              }
              performSessionRead()
            }
          } catch {
            case e => {
              complete.set(true)
              println("unexpected exception: " + e)
              assert(false)
            }
          }
        }
        complete.get
      }
    }
    
    val port = portSlot.get
    
    val clientThread = startThread {
      val writeComplete = new Slot[Boolean](None)
      val readComplete = new Slot[Boolean](None)
      var readData: String = ""
      AsyncSocket.open("localhost", port).using { clientChannel =>
        val window = Window to (new Array[Byte](100))
        def performRead(): Unit = {
          clientChannel.readSome(selector, window) { getBytesRead =>
            try {
              val bytesRead = getBytesRead()
              if (bytesRead >= 0) {
                val readString = new String(window.until(bytesRead).asString())
                if (readString.length > 0) {
                  runCount += 1
                }
                readData += readString
                if (readData.contains("Hello world") && readData.contains("Bye!")) {
                  readComplete.set(true)
                } else {
                  performRead()
                }
              } else {
                readComplete.set(true)
              }
            } catch {
              case e: ClosedChannelException => println("--> in read client channel closed ***********")
              case e: DisposedException =>
              case e: CancelledKeyException => println("--> in read client channel cancelled key **********")
              case e => {
                e.printStackTrace(System.out)
                println("unexpected exception: " + e)
                assert(false)
              }
            }
          }
        }
        performRead()
        clientChannel.writeSome(selector, "Hello world".window) { getBytesWritten =>
          try {
            val bytesWritten1 = getBytesWritten()
            requireThat { assert(bytesWritten1 == 11) }
            deadlineTimer.execute(100.millis from Time.now) {
              clientChannel.writeSome(selector, "Bye!".window) { getBytesWritten2 =>
                try {
                  val bytesWritten2 = getBytesWritten2()
                  requireThat { assert(bytesWritten2 == 4) }
                } catch {
                  case e: DisposedException => {
                    println("--> in write client channel disposed *************")
                    println("unexpected exception: " + e)
                    assert(false)
                  }
                } finally {
                  deadlineTimer.execute(100.millis from Time.now) {
                    writeComplete.set(true)
                  }
                }

              }
            }
          } catch {
            case e => {
              println("unexpected exception: " + e)
              assert(false)
            }
          }
        }
        
        writeComplete.get
        readComplete.get
      }
    }
    
    serverThread.join()
    clientThread.join()
    
    assert(runCount == 2)
    selector.dispose()
    requireThat.replay()
  }
  
  
  @Test
  def testBinaryReadWrite(): Unit = {
    import Default._
    var requireThat = new AssertionsCollector
    val selector = new AsyncSelector()
    
    val deadlineTimer = new DeadlineTimer()
    var result0 = false
    val timeline = new Timeline
    var runCount = 0
    val portSlot = new Slot[Integer](None)
    
    val serverThread = startThread {
      AsyncServerSocket.open("localhost", portRange) { completion =>
        val (serverChannel, port) = completion()
        portSlot.set(port)
        val complete = new Slot[Boolean](None)
        serverChannel.accept(selector) { getSocket =>
          try {
            for (sessionChannel <- getSocket()) {
              val window = Window to (new Array[Byte](100))
              def performSessionRead(): Unit = {
                sessionChannel.readSome(selector, window) { getBytesRead =>
                  try {
                    val bytesRead = getBytesRead()
                    if (bytesRead >= 0) {
                      sessionChannel.writeSome(selector, window.until(bytesRead)) { getBytesWritten =>
                        try {
                          val bytesWritten = getBytesWritten()
                          requireThat { assert(bytesWritten == 11 || bytesWritten == 4 || bytesWritten == 15) }
                          performSessionRead()
                        } catch {
                          case e: DisposedException => {
                            complete.set(true)
                            e.printStackTrace(System.out)
                          }
                          case e => {
                            complete.set(true)
                            e.printStackTrace(System.out)
                            println("unexpected exception: " + e)
                            assert(false)
                          }
                        }
                      }
                    } else {
                      complete.set(true)
                    }
                  } catch {
                    case e: DisposedException => {
                      println("--> in read server channel disposed *************")
                      complete.set(true)
                      e.printStackTrace(System.out)
                    }
                    case e => {
                      complete.set(true)
                      e.printStackTrace(System.out)
                      println("unexpected exception: " + e)
                      assert(false)
                    }
                  }
                }
              }
              performSessionRead()
            }
          } catch {
            case e => {
              complete.set(true)
              println("unexpected exception: " + e)
              assert(false)
            }
          }
        }
        complete.get
      }
    }
    
    val port = portSlot.get
    
    val clientThread = startThread {
      val writeComplete = new Slot[Boolean](None)
      val readComplete = new Slot[Boolean](None)
      var readData: String = ""
      AsyncSocket.open("localhost", port).using { clientChannel =>
        val stream = clientChannel.streamVia(selector)
        val window = Window to (new Array[Byte](100))
        def performRead(): Unit = {
          stream.readSome(window) { getBytesRead =>
            try {
              val bytesRead = getBytesRead()
              if (bytesRead >= 0) {
                val readString = new String(window.until(bytesRead).asString())
                if (readString.length > 0) {
                  runCount += 1
                }
                readData += readString
                if (readData.contains("Hello world") && readData.contains("Bye!")) {
                  readComplete.set(true)
                } else {
                  performRead()
                }
              } else {
                readComplete.set(true)
              }
            } catch {
              case e: ClosedChannelException => println("--> in read client channel closed ***********")
              case e: DisposedException =>
              case e: CancelledKeyException => println("--> in read client channel cancelled key **********")
              case e => {
                e.printStackTrace(System.out)
                println("unexpected exception: " + e)
                assert(false)
              }
            }
          }
        }
        performRead()
        import org.offsync.async.AsyncCombinators._
        (stream.writeCharacters("Hello world") ~ stream.writeCharacters("Bye!")) { getBytesWritten2 =>
          try {
            val bytesWritten2 = getBytesWritten2()
            println(bytesWritten2)
            assert(bytesWritten2 == (11, 4))
          } catch {
            case e: DisposedException => println("--> in write client channel disposed *************")
            case e => {
              println("unexpected exception: " + e)
              assert(false)
            }
          } finally {
            deadlineTimer.execute(100.millis from Time.now) {
              writeComplete.set(true)
            }
          }
        }
        
        writeComplete.get
        readComplete.get
      }
    }
    
    serverThread.join()
    clientThread.join()
    
    assert(runCount == 2 || runCount == 1)
    selector.dispose()
    requireThat.replay()
  }
  
  def expectException[E <: Throwable](f: => Unit)(implicit manifest: Manifest[E]): Unit = {
    try {
      f
      Assert.fail("Expected exception: \"" + manifest + "\"")
    } catch {
      case e => assert(e.isInstanceOf[E])
    }
  }
  
  @Test
  def testDispose(): Unit = {
    val selector = new AsyncSelector()
    selector.dispose()
    AsyncServerSocket.open("localhost", portRange) { completion =>
      val (listenChannel, port) = completion()
      expectException[DisposedException] {
        listenChannel.accept(selector) { completion =>
          completion()
          false
        }
      }
    }
  }
}
